package com.shujia.flink.sql

import org.apache.flink.table.api.EnvironmentSettings
import org.apache.flink.table.api.bridge.scala.StreamTableEnvironment
import org.apache.flink.streaming.api.scala._
import org.apache.flink.table.api._
import org.apache.flink.table.api.bridge.scala._
import org.apache.flink.types.Row

object Demo5TopN {

  def main(args: Array[String]): Unit = {
    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment

    val settings: EnvironmentSettings = EnvironmentSettings.newInstance()
      .useBlinkPlanner()
      .inStreamingMode()
      .build()


    val bsEnv: StreamTableEnvironment = StreamTableEnvironment.create(env, settings)


    bsEnv.executeSql(
      """
        |
        |CREATE TABLE kafkaTable (
        | word String,
        | user_action_time AS PROCTIME() -- 声明一个额外的列作为处理时间属性
        |) WITH (
        | 'connector' = 'kafka',
        | 'topic' = 'test_topic1',
        | 'properties.bootstrap.servers' = 'master:9092,node1:9092,node2:9092',
        | 'properties.group.id' = 'testGroup1',
        | 'format' = 'csv',
        | 'csv.ignore-parse-errors' = 'true',
        | 'scan.startup.mode' = 'earliest-offset'
        |)
        |
      """.stripMargin)


    //mysql建表语句
    /*

CREATE TABLE `word_topn` (
  `word` varchar(255) DEFAULT NULL,
  `end_time` datetime NOT NULL,
  `c` bigint(255) DEFAULT NULL,
  `r` bigint(255) NOT NULL,
  PRIMARY KEY (`end_time`,`r`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8;



     */



    bsEnv.executeSql(
      """
        |CREATE TABLE word_topn (
        |  word STRING,
        |  end_time TIMESTAMP(3),
        |  c bigint,
        |  r bigint,
        |  PRIMARY KEY (end_time,r) NOT ENFORCED
        |) WITH (
        |   'connector' = 'jdbc',
        |   'url' = 'jdbc:mysql://master:3306/student?useUnicode=true&characterEncoding=utf-8&useSSL=false',
        |   'table-name' = 'word_topn',
        |   'username' = 'root',
        |   'password'= '123456'
        |)
        |
      """.stripMargin)


    /**
      * 统计最新15秒单词的数量，每隔5秒统计一次，每个窗口取单词数量前2
      *
      */

    bsEnv.executeSql(
      """
        |
        |insert into word_topn
        |select * from
        |(select word,end_time,c , row_number() over(partition by end_time order by c desc) as r
        |from
        |(select
        |word,
        |HOP_END(user_action_time, INTERVAL '5' SECOND, INTERVAL '15' SECOND) as end_time,
        |count(1) as c
        |from kafkaTable
        |group by word,HOP(user_action_time, INTERVAL '5' SECOND, INTERVAL '15' SECOND)
        |) as a
        |) as b
        |where r < 3
        |
        |
      """.stripMargin)


  }

}
